package com.feriki.msync.process;

import com.feriki.msync.constants.PolicyType;
import com.feriki.msync.process.sink.ClickhouseSinkHandler;
import com.feriki.msync.process.sink.JsonLineFileSinkHandler;
import com.feriki.msync.process.sink.SinkHandler;
import com.feriki.msync.process.source.JsonLineFileSourceHandler;
import com.feriki.msync.process.source.MilvusSourceHandler;
import com.feriki.msync.process.source.SourceHandler;
import com.feriki.msync.config.Config;
import com.feriki.msync.config.ComplexConnectConfig;
import com.feriki.msync.constants.ConnectType;
import com.feriki.msync.pool.MilvusConnectionPool;
import com.feriki.msync.process.sink.MilvusSinkHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

/**
 * @author luozhida
 * @create 2024-07-01
 */
@Slf4j
public class Engine {

    private final String COVER_MODE_PROTECTION_TARGET = PolicyType.COVER.toUpperCase() + ":";
    private Config config;
    private MilvusConnectionPool sourcePool;
    private MilvusConnectionPool sinkPool;


    public Engine(Config config) {
        this.config = config;
        check(config);
    }


    public void start() {
        SourceHandler sourceHandler;
        SinkHandler sinkHandler;
        switch (this.config.getSource().getType()) {
            case ConnectType.JSONL:
                sourceHandler = new JsonLineFileSourceHandler(this.config.getSourceObjectName(),
                        this.config.getBatchSize(),
                        this.config.getLimit());
                break;
            default:
                sourcePool = createMilvusConnectionPool("sourcePool",this.config.getSource(),1);
                sourceHandler = new MilvusSourceHandler(sourcePool, this.config.getSourceObjectName(),
                        this.config.getMilvusSourceSearchExpr(),
                        this.config.getBatchSize(),
                        this.config.getLimit());
                break;
        }
        switch (this.config.getSink().getType()) {
            case ConnectType.JSONL:
                sinkHandler = new JsonLineFileSinkHandler(this.config.getSinkObjectName(), this.config.getPolicy());
                break;
            case ConnectType.CH:
                sinkHandler = new ClickhouseSinkHandler(this.config.getSinkObjectName(), this.config.getSink(),this.config.getClickhouseNumberTypeIsNullFill(),null);
                break;
            default:
                sinkPool = createMilvusConnectionPool("sinkPool",this.config.getSink(),1);
                sinkHandler = new MilvusSinkHandler(sinkPool, this.config.getSinkObjectName(),
                        this.config.getPolicy(),
                        this.config.isMilvusSinkEnableNonAutoIdGenerateValue(),
                        this.config.isMilvusSinkEnableTestPartition());
                break;
        }

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (sourcePool != null) {
                sourcePool.closeAll();
            }
            if (sinkPool != null) {
                sinkPool.closeAll();
            }
        }));

        Processor processor = new Processor(sourceHandler, sinkHandler);
        log.info("已成功下发从源{}【{}】同步数据到目标{}【{}】的任务", this.config.getSource().getType(),sourceHandler.getName(), this.config.getSink().getType(),sinkHandler.getName());
        //只有一个任务，直接运行了吧
        processor.run();
    }


    private MilvusConnectionPool createMilvusConnectionPool(String poolName, ComplexConnectConfig milvusConfig, int poolSize){
        return new MilvusConnectionPool(poolName, milvusConfig, poolSize);
    }

    private void check(Config config) {
        if (config == null) {
            throw new IllegalArgumentException("请配置正确格式的参数");
        }
        if (config.getSource() == null) {
            throw new IllegalArgumentException("请配置正确格式的源连接参数");
        }
        if (config.getSink() == null) {
            throw new IllegalArgumentException("请配置正确格式的目标连接参数");
        }
        if (ConnectType.JSONL.equals(config.getSource().getType()) &&
                ConnectType.JSONL.equals(config.getSink().getType())) {
            throw new IllegalArgumentException("不支持从文件同步到文件,请配置为\"从文件到milvus\"或\"从milvus到文件\"或\"从milvus到milvus\"");
        }
        if (StringUtils.isBlank(config.getSourceObjectName())) {
            throw new IllegalArgumentException("请配置源对象名称sourceObjectName");
        }
        if (StringUtils.isBlank(config.getSinkObjectName())) {
            throw new IllegalArgumentException("请配置目标对象名称sinkObjectName");
        }

        coverModeProtectionLock(config);
    }

    /**
     * 覆盖模式保护锁
     *
     * @author luozhida
     * @create 2024-07-12
     */
    private void coverModeProtectionLock(Config config) {
        if (config != null && config.getPolicy() != null
                && PolicyType.COVER.equalsIgnoreCase(config.getPolicy())
                && config.getSinkObjectName() != null
                && !config.getSinkObjectName().startsWith(COVER_MODE_PROTECTION_TARGET)) {
            throw new IllegalArgumentException("覆盖策略下,为防止误操作的发生,目标对象名称sinkObjectName的值必须以\""+COVER_MODE_PROTECTION_TARGET+"\"开头,请您确认并修改后再启动");
        } else {
            if (config != null && config.getSinkObjectName() != null) {
                config.setSinkObjectName(config.getSinkObjectName().replaceFirst(COVER_MODE_PROTECTION_TARGET, ""));
            }
        }
    }
}
